package cas.ihep.util;

/**
 * Created by chocolate on 2017/11/24.
 */

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hdfs.DFSClient;
import org.apache.hadoop.util.Daemon;
import org.apache.hadoop.util.Time;

import java.io.File;
import java.io.IOException;
import java.net.SocketTimeoutException;
import java.util.HashMap;
import java.util.Map;

public class RefreshFileSize {
    static final Log LOG = LogFactory.getLog(RefreshFileSize.class);
    static final long RENEWER_SLEEP_DEFAULT = 500L;
    static final long RENEWER_INTERVAL_DEFAULT = 1000L;
    private final DFSClient dfsc;


    public RefreshFileSize(final DFSClient dfsc) throws IOException {
        this.dfsc = dfsc;
    }



    private int currentId = 0;
    private Daemon daemon = null;

    private final Map<String,Map<Long,String>> files = new HashMap<String,Map<Long,String>>();

    synchronized public void put(final String src, final long fileId, final String path) {
        if (!isRunning()) {
            //start a new deamon with a new id.
            final int id = ++currentId;
            daemon = new Daemon(new Runnable() {
                @Override
                public void run() {
                    try {
                        RefreshFileSize.this.run(id);
                    } catch(InterruptedException e) {
                        LOG.info("ERROR:"+e.getMessage());
                    }
                }});
            daemon.start();
        }
        addFile(src,fileId,path);
    }

    /**
     private synchronized long getSleepPeriod() {
     long sleepPeriod = RENEWER_SLEEP_DEFAULT;
     return sleepPeriod;
     }
     */
    /** @return the renewal time in milliseconds. */
    /**
     private synchronized long getRenewalTime() {
     long renewal = RENEWER_INTERVAL_DEFAULT;
     return renewal;
     }
     */
    /** Is the daemon running? */
    synchronized boolean isRunning() {
        return daemon != null && daemon.isAlive();
    }
    /** Add a file. */
    private synchronized void addFile(final String src, final long fileId, final String path) {
        if(files.containsKey(src)) {
            return;
        }
        Map<Long,String> map = new HashMap<Long,String>();
        map.put(fileId,path);
        files.put(src,map);
        LOG.info("files size : "+files.size());
    }
    /**
     * Periodically check in with the namenode and renew all the leases
     * when the lease period is half over.
     */
    private void run(final int id) throws InterruptedException {
        for(long lastRenewed = Time.now(); !Thread.interrupted();
            Thread.sleep(RENEWER_SLEEP_DEFAULT)) {
            final long elapsed = Time.now() - lastRenewed;
            if (elapsed >= RENEWER_INTERVAL_DEFAULT) {
                try {
                    renew();
                    lastRenewed = Time.now();
                } catch (SocketTimeoutException ie) {
                    break;
                } catch (IOException ie) {
                    LOG.warn(ie.getMessage());
                }
            }

            synchronized(this) {
                if (id != currentId ) {
                    //no longer the current daemon or expired
                    return;
                }


            }
        }
    }

    private void renew() throws IOException {
        final Map<String,Map<Long,String>> copies;
        synchronized(this) {
            copies = new HashMap<String,Map<Long,String>>(files);
        }
        String previousFile = "";
        for (String src : copies.keySet()) {
            if (!src.equals(previousFile)) {
                for(long id : copies.get(src).keySet()){
                    String path = copies.get(src).get(id);
                    File file = new File(path);
                    if(file.exists()){
                        long size = file.length();
                        dfsc.getNamenode().fsync(src, id, dfsc.getClientName(), size);
                    }
                }

            }
            previousFile = src;
        }
    }
    public void remove(String src) throws IOException {
        files.remove(src);
        LOG.info("files size : "+files.size());
        if(files.size() < 1 && daemon.isAlive()){
            daemon.interrupt();
        }
    }
    public void close() throws IOException {
        files.clear();
        if(daemon.isAlive()){
            daemon.interrupt();
        }
    }

}

